package com.neu;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.*;
import java.net.URI;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author ：lychallengers@gmail.com
 * @date ：2021/1/9 12:28
 */
public class DBServiceImpl  implements DBService, Serializable {
    private static final long serialVersionUID = -6265623952497030441L;
    static final String JDBC_DRIVER = "com.omnisci.jdbc.OmniSciDriver";
    static final String DB_URL_PRE = "jdbc:omnisci:";
    static final String DB_URL_AFTER=":6274:omnisci";
    static final String USER = "admin";
    static final String PASS = "HyperInteractive";
    static final String CONFIG_PATH="setting.properties";
    public String RemotePath;
    public String LocalPath;
    public String BufferSize;
    public String ZkServers;
    public String KafkaServers;
    public String MsfUrl;
    Connection conn = null;
    Statement stmt = null;
    FileSystem fileSystem=null;
    Properties props=null;//获取配置文件
    List<String> columnName;
    List<List<String>> result;
    double exeTime;

    public boolean initDB() {
        try {
            props = new Properties();
            props.load(new java.io.FileInputStream(CONFIG_PATH));
            String server_ip = props.getProperty("server_ip", "219.216.65.81");
            RemotePath = props.getProperty("remote_path", "/home/neu");
            LocalPath = props.getProperty("local_path", "./");
            BufferSize = props.getProperty("bufferSize", "2048");
            KafkaServers = props.getProperty("kafka_servers",
                    "219.216.65.160:9092,219.216.65.161:9092,219.216.65.81:9092");
            ZkServers = props.getProperty("zk_servers",
                    "219.216.65.160:2181,219.216.65.161:2181,219.216.65.81:2181");
            MsfUrl = props.getProperty("msf_url", "mfs://219.216.65.160:8888");
            Configuration configuration = new Configuration();
            configuration.set("bufferSize", BufferSize);//buffersize?
            configuration.set("zkServers", ZkServers);
            configuration.set("kafkaServers", KafkaServers);
            fileSystem = FileSystem.get(new URI(MsfUrl), configuration);  //如用户无权限，可加用户参数
            System.out.println("initDataBase Success\n");
            Class.forName(JDBC_DRIVER);
            conn = DriverManager.getConnection(DB_URL_PRE + server_ip + DB_URL_AFTER, USER, PASS);
            stmt = conn.createStatement();
            return true;
        }
        catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    public String getFirstColumnValue(String sql) {
        try{
            String result=null;
            Pattern pattern = Pattern.compile("(?<=\\()(.+?)(?=\\))");
            String str = sql;
            Matcher matcher = pattern.matcher(str);
            boolean is = matcher.find();
            if (is) {
                String s = matcher.group();
                String[] arr = s.trim().split(",");
                String column_value = arr[0].trim();
//                if (column_value.endsWith("'")) {
//                    column_value = column_value.substring(1, column_value.length() - 1);
//                }
                result=column_value;
            }
            return result;
        }
        catch (Exception e){
            e.printStackTrace();
            return null;
        }
    }

    public String getTableName(String sql){
        try{
            String [] res=sql.split("\\s+");
            if(res.length<3){
                return null;
            }
            else{
                return res[2];
            }
        }
        catch (Exception e){
            e.printStackTrace();
            return null;
        }

    }
    public String[] hasKey(Statement stmt,String sql) {
        try{
            String firstColumnValue=this.getFirstColumnValue(sql);
            if(firstColumnValue==null)return new String[0];
            String tableName=this.getTableName(sql);
            if(tableName==null)return new String[0];
            String querySql="select * from "+tableName+" limit 1";
            ResultSet resultSet=stmt.executeQuery(querySql);
            String firstCloumnName=resultSet.getMetaData().getColumnLabel(1);
            String newQuerySql="select * from "+tableName+" where "+firstCloumnName+"="+firstColumnValue;
            resultSet=stmt.executeQuery(newQuerySql);
            if(resultSet.next()){
                return new String[]{tableName,firstCloumnName,firstColumnValue};
            }
            return  new String[0];
        }catch (Exception e){
            e.printStackTrace();
            return new String[0];
        }
    }
    public String convertGetToSelect(String sql){
        try{
            String [] re=sql.split("\\s+");
            StringBuffer sb=new StringBuffer();
            if(re.length<4)return null;
            String s1="select * ";
            String s2=" where ";
            sb.append(s1);sb.append(" from ");
            sb.append(re[re.length-1]);
            sb.append(s2+" ");
            for(int i=1;i<re.length-2;i++){
                sb.append(re[i]);
            }
            return  sb.toString();
        }
        catch (Exception e){
            e.printStackTrace();
            return null;
        }

    }

    public boolean deleteKey(Statement stmt, String sql,String [] sendData){
        try{
            String tableName=sendData[0];
            String firstCloumnName=sendData[1];
            String firstColumnValue=sendData[2];
            String deleteSt="delete from "+tableName+" where "+firstCloumnName+" =  "+firstColumnValue;
            stmt.executeUpdate(deleteSt);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }

    public String pathTransform(String originPath){
        int t1=Math.abs(originPath.hashCode()%7);
        int t2=Math.abs(originPath.hashCode()%5);
        return "/0"+"/"+Integer.toString(t2)+"/"+originPath;
    }
    //读取文件
    public String getFile(FileSystem fileSystem,String getPath){
        try{
            long beginTime = System.nanoTime();
            Path path = new Path(pathTransform(getPath));//转为远程目录
            InputStream inputStream = fileSystem.open(path);

            //output stream
            File file = new File(getPath);
            OutputStream outputStream= new FileOutputStream(file);
            //bytecount
            int bytesWritten = 0;
            int byteCount = 0;
            //byte count
            byte[] bytes = new byte[10240];//buffer size
            while ((byteCount = inputStream.read(bytes)) != -1)
            {
                outputStream.write(bytes, bytesWritten, byteCount);
                bytesWritten += byteCount;
            }
            long endTime = System.nanoTime();
            long costTime = (endTime - beginTime)/1000;
            exeTime=(double)costTime/1000;
            return "Success write file to local file";
        } catch (Exception e) {
            e.printStackTrace();
            return "error";
        }
    }

    //写入文件(远程)
    public String putFile(FileSystem fileSystem,String putPath) {
        try{
            long beginTime = System.nanoTime();
            //output stream remote file
            Path path = new Path(pathTransform(putPath));
            OutputStream fsDataOutputStream = fileSystem.create(path);
            //input stream local file
            InputStream inputStream = new FileInputStream(putPath);
            //bytecount
            int bytesWritten = 0;
            int byteCount;
            //byte count
            byte[] bytes = new byte[10240];
            while ((byteCount = inputStream.read(bytes)) != -1)
            {
                fsDataOutputStream.write(bytes, bytesWritten, byteCount);
                bytesWritten += byteCount;
            }
            fsDataOutputStream.close();
            long endTime = System.nanoTime();
            long costTime = (endTime - beginTime)/1000;
            exeTime=(double)costTime/1000;
            return "Success write file to remote";
        } catch (Exception e) {
            e.printStackTrace();
            return "error";
        }

    }

    //删除文件(远程)
    public String delFile(FileSystem fileSystem,String delPath) {
        try{
            long beginTime = System.nanoTime();
            //output stream remote file
            Path path = new Path(pathTransform(delPath));
            fileSystem.delete(path,true);
            long endTime = System.nanoTime();
            long costTime = (endTime - beginTime)/1000;
            exeTime=(double)costTime/1000;
            return "Success delete remote file";
        } catch (Exception e) {
            e.printStackTrace();
            return "error";
        }

    }


    public String getGraph(String sql) {
        try {
            String[] cmd_in = sql.split(" ");

            String cmd = "/root/miniconda3/bin/python /neu-storage/neugraph/client.py " + cmd_in[0] + " " + cmd_in[1] + " " + cmd_in[2];

            System.out.println(cmd);

            Process pr = Runtime.getRuntime().exec(cmd);
            BufferedReader in = new BufferedReader(new InputStreamReader(pr.getInputStream()));
            String line2;
            while ((line2 = in.readLine()) != null) {
                System.out.println(line2);
            }
            in.close();
            pr.waitFor();


            File file = new File("/neu-storage/neugraph/out.txt");
            BufferedReader reader = null;
            String tempString = null;

            String result = "";
            reader = new BufferedReader(new FileReader(file));
            while ((tempString = reader.readLine()) != null) {
                result += tempString + "\n";
            }
            reader.close();
            return result;
            // System.out.print(result);
        }catch (Exception e){
            e.printStackTrace();
            return null;
        }

    }

    /**(
     * 0 代表语句正常，sql执行正常
     * -1 代表执行错误
     *
     * @param sql
     * @return
     */
    public DBResult exeSql(String sql) {
        try{
            initDB();
            exeTime=-1;
            sql=sql.replaceAll("\n"," ");
            sql=sql.replaceAll("\r"," ");
            sql=sql.replaceAll("\t"," ").trim();
            if(sql.endsWith(";")){
                sql=sql.substring(0,sql.length()-1);
            }
            String [] tmp=sql.split("\\s+");
            if(tmp.length==0)return new DBResult(-1);

            if(tmp[0].compareToIgnoreCase("pagerank")==0||tmp[0].compareToIgnoreCase("sssp")==0
                    ||tmp[0].compareToIgnoreCase("load")==0||tmp[0].compareToIgnoreCase("edge")==0
                    ||tmp[0].compareToIgnoreCase("info")==0||tmp[0].compareToIgnoreCase("hops")==0
                    ||tmp[0].compareToIgnoreCase("neighbor")==0||tmp[0].compareToIgnoreCase("node")==0||
                    tmp[0].compareToIgnoreCase("path")==0){
                long beginTime = System.nanoTime();
                String graphResult= getGraph(sql);
                long endTime = System.nanoTime();
                long costTime = (endTime - beginTime)/1000;
                List<String> graphColumnName=new ArrayList<>();
                graphColumnName.add("graph info");
                List<List<String>> graphRes=new ArrayList<>();
                graphRes.add(new ArrayList<>());
                graphRes.get(0).add(graphResult);
                return new DBResult(0,graphColumnName,graphRes,costTime);
            }

            //先判断是否为 getf,putf
            if(tmp[0].compareToIgnoreCase("getf")==0){
                if(tmp.length==1){
                    return new DBResult(-1);
                }
                else{
                    getFile(fileSystem,tmp[1]);
                    return new DBResult(0,exeTime);
                }
            }
            else if(tmp[0].compareToIgnoreCase("putf")==0){
                if(tmp.length==1){
                    return new DBResult(-1);
                }
                else{
                    putFile(fileSystem,tmp[1]);
                    return new DBResult(0,exeTime);
                }
            }
            else if(tmp[0].compareToIgnoreCase("delf")==0){
                if(tmp.length==1){
                    return new DBResult(-1);
                }
                else{
                    delFile(fileSystem,tmp[1]);
                    return new DBResult(0,exeTime);
                }
            }

            if(tmp[0].compareToIgnoreCase("select")==0||
                    tmp[0].compareToIgnoreCase("show")==0||
                    tmp[0].compareToIgnoreCase("get")==0){//执行query
                if(tmp[0].compareToIgnoreCase("get")==0){
                    sql=this.convertGetToSelect(sql);
                }
                //计算执行时间
                long beginTime = System.nanoTime();
                ResultSet resultSet=stmt.executeQuery(sql);
                long endTime = System.nanoTime();
                long costTime = (endTime - beginTime)/1000;
                exeTime=(double)costTime/1000;

                columnName=new ArrayList<>();
                for(int i=0;i<resultSet.getMetaData().getColumnCount();i++){
                    columnName.add(resultSet.getMetaData().getColumnLabel(i+1));
                }
                result=new ArrayList<>();
                while(resultSet.next()){
                    List<String> tmpList=new ArrayList<>();
                    for(int i=0;i<columnName.size();i++){
                        String t=resultSet.getString(i+1);
                        tmpList.add(t);
                    }
                    result.add(tmpList);
                }
                return new DBResult(0,columnName,result,exeTime);
            }
            else{
                if(tmp[0].compareToIgnoreCase("insert")==0){
                    if(this.hasKey(stmt,sql).length!=0){
                        return new DBResult(-1);
                    }
                }
                else if(tmp[0].compareToIgnoreCase("put")==0){
                    String [] keyArr=this.hasKey(stmt,sql);
                    if(keyArr.length!=0){//need delete
                        if(!this.deleteKey(stmt,sql,keyArr)){
                            return new DBResult(-1);
                        }
                    }
                    sql=sql.replaceFirst(tmp[0],"insert");
                }
                else if(tmp[0].compareToIgnoreCase("del")==0){
                    sql=sql.replaceFirst(tmp[0],"delete");
                }
                long beginTime = System.nanoTime();
                int r=stmt.executeUpdate(sql);
                long endTime = System.nanoTime();
                long costTime = (endTime - beginTime)/1000;
                exeTime=(double)costTime/1000;
                return new DBResult(r,exeTime);
            }
        } catch (Exception e){
            e.printStackTrace();
            return new DBResult(-1);
        }
    }



}
